package cn.doitedu.api;

import beans.UserAction;
import com.alibaba.fastjson.JSON;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


public class _13_Exercise_ProcessFunction_Timer {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8899);

        SingleOutputStreamOperator<UserAction> beans = stream.map(json -> JSON.parseObject(json, UserAction.class));

        beans.keyBy(UserAction::getUid)
                .process(new KeyedProcessFunction<Integer, UserAction, String>() {

                    @Override
                    public void processElement(UserAction value, KeyedProcessFunction<Integer, UserAction, String>.Context ctx, Collector<String> out) throws Exception {

                        out.collect("hello "+ value.getUid());

                        // 注册定时器
                        ctx.timerService().registerProcessingTimeTimer(System.currentTimeMillis() + 5000);

                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<Integer, UserAction, String>.OnTimerContext ctx, Collector<String> out) throws Exception {

                        out.collect("hello "+ ctx.getCurrentKey());


                        ctx.timerService().registerProcessingTimeTimer(System.currentTimeMillis() + 5000);


                    }
                }).print();



        env.execute();


    }

}
